package com.seventeen.monster.mq.proxy.server.proxy.producer;

import com.alibaba.fastjson.JSON;
import com.seventeen.monster.common.constant.DelFlagConstants;
import com.seventeen.monster.common.constant.InstanceTypeConstants;
import com.seventeen.monster.common.exception.base.BaseException;
import com.seventeen.monster.mq.proxy.server.proxy.ext.DefaultExtMqProducer;
import com.seventeen.monster.mq.proxy.server.proxy.ext.NewExtMqProducer;
import com.seventeen.monster.mq.proxy.server.proxy.thread.ProducerPool;
import com.seventeen.monster.mq.proxy.domain.MqInstance;
import com.seventeen.monster.mq.proxy.domain.MqProducerProxy;
import com.seventeen.monster.mq.proxy.service.IMqInstanceService;
import com.seventeen.monster.mq.proxy.service.IMqProducerProxyService;
import com.seventeen.monster.mq.proxy.server.proxy.ext.BaseExtMqProducer;
import com.seventeen.monster.mq.proxy.server.proxy.ext.ExtSendMessage;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 生产者代理工厂
 *
 * @author seventeen
 */
@Component
public class ProducerFactory implements InitializingBean, DisposableBean {
    public static final Logger LOGGER = LoggerFactory.getLogger(ProducerFactory.class);
    private final ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(1,
            new BasicThreadFactory.Builder().namingPattern("producer-factory-schedule-pool-%d").daemon(true).build());
    private final Map<String, Long> producerIdMap = new ConcurrentHashMap<>(16);
    @Autowired
    private IMqInstanceService mqInstanceService;
    @Autowired
    private IMqProducerProxyService mqProducerProxyService;

    private ProducerPool producerPool;

    public BaseExtMqProducer getProducer(String producerGroupName) {
        Long producerProxyId = producerIdMap.get(producerGroupName);
        if (producerProxyId == null) {
            return null;
        }
        return producerPool.get(producerProxyId);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        producerPool = new ProducerPool();
        producerPool.setDaemon(true);
        producerPool.start();
        this.scheduledExecutorService.scheduleAtFixedRate(() -> {
            try {
                List<MqProducerProxy> producerProxyList = mqProducerProxyService.list();
                Set<String> set = new HashSet<>();
                for (MqProducerProxy mqProducerProxy : producerProxyList) {
                    addProducerProxy(mqProducerProxy);
                    if (DelFlagConstants.NORMAL.equals(mqProducerProxy.getDelFlag())) {
                        set.add(mqProducerProxy.getProducerProxyGroup());
                        producerIdMap.put(mqProducerProxy.getProducerProxyGroup(), mqProducerProxy.getId());
                    }
                }

                for (String key : producerIdMap.keySet()) {
                    if (!set.contains(key)) {
                        producerIdMap.remove(key);
                    }
                }

            } catch (Exception e) {
                LOGGER.error(e.getMessage(), e);
            }
        }, 1000 * 2, 1000 * 60, TimeUnit.MILLISECONDS);
    }

    @Override
    public void destroy() throws Exception {
        LOGGER.info("producerFactory destroy!");
        producerPool.toStop();
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            LOGGER.warn("", e);
        }
    }

    private void addProducerProxy(MqProducerProxy mqProducerProxy) {
        MqInstance mqInstance = mqInstanceService.selectById(mqProducerProxy.getInstanceId());
        BaseExtMqProducer producer = getExtMqProducerByInstanceType(mqInstance.getInstanceType());
        producer.setId(mqProducerProxy.getId());
        producer.setProducerProxyGroup(mqProducerProxy.getProducerProxyGroup());
        producer.setVersion(mqProducerProxy.getVersion());
        producer.setDelFlag(mqProducerProxy.getDelFlag());
        producer.setAccessKey(mqInstance.getAccessKey());
        producer.setSecretKey(mqInstance.getSecretKey());
        producer.setEndpoints(mqInstance.getEndpoints());
        producer.setAccessChannel(mqInstance.getAccessChannel());
        if (StringUtils.isNotBlank(mqProducerProxy.getWithTopics())) {
            Set<String> withTopics = new HashSet<>(Arrays.asList(mqProducerProxy.getWithTopics().split(",")));
            producer.setWithTopics(withTopics);
        }
        producerPool.pushQueue(producer);
    }

    private BaseExtMqProducer getExtMqProducerByInstanceType(String instanceType) {

        BaseExtMqProducer extMqProducer;
        switch (instanceType) {
            case InstanceTypeConstants.V4_X:
            case InstanceTypeConstants.V_ALI_CLOUD_4_X:
                extMqProducer = new DefaultExtMqProducer();
                break;
            case InstanceTypeConstants.V5_X:
            case InstanceTypeConstants.V_ALI_CLOUD_5_X:
                extMqProducer = new NewExtMqProducer();
                break;
            default:
                extMqProducer = null;
        }
        return extMqProducer;
    }

    public void sendSync(ExtSendMessage sendMessage) {
        LOGGER.info("send message start, sendMessage {} ", JSON.toJSONString(sendMessage));
        BaseExtMqProducer mqProducer = null;
        try {
            mqProducer = getProducer(sendMessage.getProducerGroupName());
        } catch (Exception e) {
            LOGGER.error("get producer {} fail", sendMessage.getProducerGroupName(), e);
        }
        if (Objects.isNull(mqProducer)) {
            throw new BaseException("生产者代理不存在");
        }
        mqProducer.sendMsg(sendMessage);
    }

}
